<?php
require "././vendor/autoload.php";

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Exchange\AMQPExchangeType;
use PhpAmqpLib\Message\AMQPMessage;

$exchangeName = "publishExchange";
$queueName = "publishQueueName";


$host = PHP_OS == "Linux" ? "127.0.0.1" : "10.0.0.200";
$conn = new AMQPStreamConnection($host, '5672', 'wyqgg', '123456');

$ch = $conn->channel();

$ch->confirm_select();

$ch->set_ack_handler(function (AMQPMessage $message) {
    echo "消息传递成功" . $message->getBody() . PHP_EOL;
});

$ch->set_nack_handler(function (AMQPMessage $message) {
    echo "消息传递失败" . $message->getBody() . PHP_EOL;
});

$ch->set_return_listener(function ($replyCode, $replyText, $exchangeName, $routeKey, AMQPMessage $message) {
    echo "消息投递失败错误码:" . $replyCode . PHP_EOL .
        "错误信息:" . $replyText . PHP_EOL .
        "交换机:" . $exchangeName . PHP_EOL .
        "路由键:" . $routeKey . PHP_EOL .
        "消息主体:" . $message->getBody() . PHP_EOL;
});

$ch->exchange_declare($exchangeName, AMQPExchangeType::FANOUT);
$ch->queue_declare($queueName, false, true);
$ch->queue_bind($queueName, $exchangeName);
//生成消息

for ($i=1;$i<1000;$i++){
    $messageBody = "publish".$i;
    $pro = ['content_type' => "text/plain"];
    
    $message = new AMQPMessage($messageBody, $pro);

    $ch->basic_publish($message, $exchangeName,'', true);

    $ch->wait_for_pending_acks_returns();
}

$ch->close();
$conn->close();